package spring.cloud.tasks.executor.executor.task;

import com.alibaba.fastjson2.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.TaskType;
import spring.cloud.tasks.common.Threads;
import spring.cloud.tasks.common.task_item_allot.TaskItem;
import spring.cloud.tasks.executor.executor.task.execute_runtime_context.*;
import spring.cloud.tasks.executor.executor.task.task_item_execute.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
@Setter
@Getter
public class TaskExecuteEngine {
    public volatile boolean needStop = false;//外部尝试关闭任务执行器
    //
    protected String nameSpace;
    protected String executorId;
    protected String taskId;
    protected TaskExecutor taskExecutor;

    protected TaskItemExecuteRenewalInfoService taskItemExecuteInfoService;
    //
    private volatile boolean stopped = false;
    private volatile boolean forceStopped = false;
    private volatile boolean running = false;


    public final void startTaskItemExecuteThread() {
        if (isNeedStop()) {
            log.warn("暂停任务执行:" + taskId);
            return;
        }

        try {
            running = true;
            TaskExecuteContext taskExecuteContext = getTaskExecuteContext();
            if (taskExecuteContext.getTaskItemExecuteRuntimeContextList().isEmpty()) {
                log.warn("任务分片项数为空 请检查");
                return;
            }
            log.info("开始执行任务:" + taskId + ",任务执行信息:" + JSON.toJSONString(taskExecuteContext));
            startTaskItemExecuteThread(taskExecuteContext);
            log.info("开执行任务:" + taskId + "完成,任务执行信息:" + JSON.toJSONString(taskExecuteContext));
        } catch (Exception e) {
            log.info(e.getMessage(), e);
        } finally {
            running = false;
        }
    }

    /**
     * 获取当前任务服务器运行时分片上下文.计算出当前能够运行的任务片
     *
     * @return 当前任务服务器运行时分片上下文
     */
    public TaskExecuteContext getTaskExecuteContext() {
        TaskExecuteContext taskExecuteContext = new TaskExecuteContext();
        taskExecuteContext.setNameSpace(nameSpace);
        taskExecuteContext.setExecutorId(executorId);
        taskExecuteContext.setTaskId(taskId);

        taskExecuteContext.setTaskConfiguration(taskExecutor.getTaskConfiguration());
        taskExecuteContext.setTaskType(taskExecuteContext.getTaskConfiguration().getTaskBaseInfo().getTaskType());
        List<TaskItemExecuteRuntimeContext> list = getNeedRunningTaskItemExecuteRuntimeContextList(taskExecuteContext);
        taskExecuteContext.setTaskItemExecuteRuntimeContextList(list);
        if (list.size() == 0) {
            return taskExecuteContext;
        }//end
        taskExecuteContext.setTaskProperties(taskExecutor.getTaskConfiguration().reloadTaskBaseInfoDto().getTaskProperties());
        taskExecuteContext.setTaskDataRangeRule(taskExecutor.getTaskConfiguration().reloadTaskBaseInfoDto().getTaskDataRangeRule());
        taskExecuteContext.setTaskItemInfo(taskExecutor.getTaskConfiguration().reloadTaskBaseInfoDto().getTaskItemInfo());
        //
        taskExecuteContext.setSecondAfterTaskStopCommandInstructed(taskExecutor.getTaskConfiguration().getSecondAfterTaskStopCommandInstructed());
        if (taskExecutor.getTaskConfiguration().reloadTaskBaseInfoDto().getTimeoutMinutes() > 0) {
            taskExecuteContext.setTimeoutMinutes(taskExecutor.getTaskConfiguration().reloadTaskBaseInfoDto().getTimeoutMinutes());
        }
        //
        if (taskExecuteContext.getTaskType().equals(TaskType.SIMPLE_SLICING)) {
            Map<Integer, String> taskItemIndexToTaskItemParameterMap = taskExecutor.getTaskConfiguration().getTaskItemIndexToTaskItemParameterMap(taskExecuteContext.getTaskType(), taskExecuteContext.getTaskItemInfo());
            taskExecuteContext.getTaskItemIndexToTaskItemParameterMap().putAll(taskItemIndexToTaskItemParameterMap);
        }
        return taskExecuteContext;
    }

    /**
     * 允许被分配的任务分片
     */
    private List<TaskItemExecuteRuntimeContext> getNeedRunningTaskItemExecuteRuntimeContextList(TaskExecuteContext taskExecuteContext) {
        List<TaskItemExecuteRuntimeContext> list = new ArrayList<TaskItemExecuteRuntimeContext>();
        //进行预分配都是进行整体任务分配下发。
        ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentHashMap<String, TaskItem>>>> taskTaskItemMap = taskExecutor.getExecutor().getTaskMap().getTaskTaskItemMap(taskId);
        if (taskTaskItemMap == null) {//任务
            return list;
        }
        //
        for (String taskInstanceId : taskTaskItemMap.keySet()) {
            ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentHashMap<String, TaskItem>>> b = taskTaskItemMap.get(taskInstanceId);
            if (b == null) {//实例
                continue;
            }
            for (String taskStage : b.keySet()) {
                ConcurrentHashMap<String, ConcurrentHashMap<String, TaskItem>> c = b.get(taskStage);
                if (c == null) {//taskStage
                    continue;
                }
                for (String taskStage$type : c.keySet()) {
                    ConcurrentHashMap<String, TaskItem> d = c.get(taskStage$type);
                    if (d == null) {//taskStage$type
                        continue;
                    }
                    Set<String> taskItemIndexSet = d.keySet();
                    for (String taskStage$type$index : taskItemIndexSet) {
                        TaskItem taskItem = d.get(taskStage$type$index);
                        if (taskItem != null) {
                            fixTaskItemExecuteRuntimeInfo(taskId, taskInstanceId, taskStage, taskStage$type, Integer.parseInt(taskStage$type$index), taskItem.getNextFireDate(), taskItem.getTimeoutDateTime(), taskItem.getLockDateTime(), taskExecuteContext, list);
                        }
                    }
                }
            }

        }
        return list;
    }

    /**
     * 可能任务已经在跑了 那么就不进行添加了
     */
    private void fixTaskItemExecuteRuntimeInfo(
            String taskId,
            String taskInstanceId,
            String taskStage,
            String taskStage$type,
            int taskStage$type$index,
            Date nextFireDate, //
            Date timeoutDateTime, //
            Date lockDateTime, //
            TaskExecuteContext taskExecuteContext, //
            List<TaskItemExecuteRuntimeContext> list//
    ) {
        TaskExecuteRuntimeContext taskExecuteRuntimeContext = TasksExecuteRuntimeContext.getTaskExecuteRuntimeContext(taskId);
        TaskInstanceExecuteRuntimeContext taskInstanceExecuteRuntimeContext = taskExecuteRuntimeContext.getTaskInstanceExecuteRuntimeContext(taskInstanceId);
        //上面两个对象必然有值
        ConcurrentHashMap<String, TaskStageExecuteRuntimeContext> taskStageToTaskStageExecuteRuntimeContextMap = taskInstanceExecuteRuntimeContext.taskStageToTaskStageExecuteRuntimeContextMap;
        TaskStageExecuteRuntimeContext taskStageExecuteRuntimeContext = taskStageToTaskStageExecuteRuntimeContextMap.get(taskStage);
        if (taskStageExecuteRuntimeContext == null) {
            taskStageExecuteRuntimeContext = new TaskStageExecuteRuntimeContext();
            taskStageToTaskStageExecuteRuntimeContextMap.put(taskStage, taskStageExecuteRuntimeContext);
        }
        //
        ConcurrentHashMap<String, TaskStage$typeExecuteRuntimeContext> taskStage$typeToTaskStage$typeExecuteRuntimeContextMap = taskStageExecuteRuntimeContext.taskStage$typeToTaskStage$typeExecuteRuntimeContextMap;
        TaskStage$typeExecuteRuntimeContext taskStage$typeExecuteRuntimeContext = taskStage$typeToTaskStage$typeExecuteRuntimeContextMap.get(taskStage$type);
        if (taskStage$typeExecuteRuntimeContext == null) {
            taskStage$typeExecuteRuntimeContext = new TaskStage$typeExecuteRuntimeContext();
            taskStage$typeToTaskStage$typeExecuteRuntimeContextMap.put(taskStage$type, taskStage$typeExecuteRuntimeContext);
        }
        ConcurrentHashMap<String, TaskItemExecuteRuntimeContext> taskItemIndexToTaskItemExecuteRuntimeContextMap = taskStage$typeExecuteRuntimeContext.taskItemIndexToTaskItemExecuteRuntimeContextMap;
        TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext = taskItemIndexToTaskItemExecuteRuntimeContextMap.get(String.valueOf(taskStage$type$index));
        //
        String taskItemId = TaskItemExecuteRuntimeContext.getTaskItemId(taskId, taskInstanceId, taskStage, taskStage$type, taskStage$type$index);
        if (taskItemExecuteRuntimeContext == null) {
            //这个地方是创建任务的入口
            taskItemExecuteRuntimeContext = new TaskItemExecuteRuntimeContext(//
                    taskExecuteContext.getNameSpace(), //
                    taskExecuteContext.getExecutorId(), //
                    taskExecuteContext.getTaskId(),//
                    taskInstanceId, //
                    taskStage,
                    taskStage$type,
                    taskStage$type$index,//
                    nextFireDate,//
                    timeoutDateTime,//
                    lockDateTime,//
                    taskExecuteContext.getMaxAllowRunningTimeMinutes());
            /**
             * 同一个时间点只会运行一个实例的一个阶段。如果开启了新的阶段，历史的阶段需要清理掉。
             * 1如果不能清理，那么批次任务业务控制逻辑需要给个标志，告诉执行业务线程不能进行处理了。
             * 2如果不能清理，调度框架需要保证
             */
            taskItemIndexToTaskItemExecuteRuntimeContextMap.put(taskItemId, taskItemExecuteRuntimeContext);
            list.add(taskItemExecuteRuntimeContext);
            return;
        } else {
            //检查当前任务分片是否完成。如果完成了 那么可以在完成后一段时间进行删除。
            if (taskItemExecuteRuntimeContext.isFinished()) {
                //是否满足删除条件
                Long allowDeleteTaskItemExecuteRuntimeContextTimeStamp = taskItemExecuteRuntimeContext.getAllowDeleteTaskItemExecuteRuntimeContextTimeStamp();
                if (allowDeleteTaskItemExecuteRuntimeContextTimeStamp == null) {
                    log.warn("可允许删除时间点为空");
                    taskItemIndexToTaskItemExecuteRuntimeContextMap.remove(String.valueOf(taskStage$type$index));
                } else {
                    if (System.currentTimeMillis() > allowDeleteTaskItemExecuteRuntimeContextTimeStamp) {
                        taskItemIndexToTaskItemExecuteRuntimeContextMap.remove(String.valueOf(taskStage$type$index));
                    } else {
                        //
                    }
                }
            } else {
                //没有完成那么该任务信息不能删除
            }
        }
    }


    protected final void startTaskItemExecuteThread(final TaskExecuteContext taskExecuteContext) {
        List<TaskItemExecuteRuntimeContext> list = new ArrayList<>();//需要执行的
        //
        List<TaskItemExecuteRuntimeContext> newTaskItemExecuteRuntimeContextList = taskExecuteContext.getTaskItemExecuteRuntimeContextList();
        for (TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext : newTaskItemExecuteRuntimeContextList) {
            taskItemExecuteInfoService.registerTaskItemExecuteRenewalInfo(taskItemExecuteRuntimeContext);
        }
        //需要搜集那些任务分片没有续期成功，需要等下次调度的时候进行续期。
        List<TaskItemExecuteRuntimeContext> oldNeedReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList = new ArrayList<>(TasksExecuteRuntimeContext.getTaskExecuteRuntimeContext(taskId).needReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList);
        for (TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext : oldNeedReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList) {
            taskItemExecuteInfoService.registerTaskItemExecuteRenewalInfo(taskItemExecuteRuntimeContext);
        }
        //
        List<TaskItemExecuteRuntimeContext> newNeedReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList = new ArrayList<>();
        for (TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext : newTaskItemExecuteRuntimeContextList) {
            if (taskItemExecuteRuntimeContext.isNeedReRenewalAtNextExecutePeriod) {
                taskItemExecuteRuntimeContext.isNeedReRenewalAtNextExecutePeriod = false;//init
                newNeedReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList.add(taskItemExecuteRuntimeContext);
            } else {
                list.add(taskItemExecuteRuntimeContext);
            }
        }
        TasksExecuteRuntimeContext.getTaskExecuteRuntimeContext(taskId).needReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList.addAll(newNeedReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList);
        //
        for (TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext : oldNeedReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList) {
            if (!taskItemExecuteRuntimeContext.isNeedReRenewalAtNextExecutePeriod) {
                TasksExecuteRuntimeContext.getTaskExecuteRuntimeContext(taskId).needReRenewalAtNextExecutePeriodTaskItemExecuteRuntimeContextList.remove(taskItemExecuteRuntimeContext);
                list.add(taskItemExecuteRuntimeContext);
            }
        }
        //
        final Integer secondAfterTaskStopCommandInstructed = taskExecuteContext.getSecondAfterTaskStopCommandInstructed();
        final Integer timeoutMinutes = taskExecuteContext.getTimeoutMinutes();
        List<TaskItemExecute> taskItemExecuteList = new ArrayList<TaskItemExecute>();
        List<TaskItemExecuteTask> taskItemExecuteTaskList = new ArrayList<TaskItemExecuteTask>();
        for (TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext : list) {
            String taskId = taskItemExecuteRuntimeContext.getTaskId();
            String taskInstanceId = taskItemExecuteRuntimeContext.getTaskInstanceId();
            String taskStage = taskItemExecuteRuntimeContext.getTaskStage();
            String taskStage$type = taskItemExecuteRuntimeContext.getTaskStage$type();
            Integer taskStage$type$index = taskItemExecuteRuntimeContext.getTaskStage$type$index();
            //
            String taskDataRangeRule = taskItemExecuteRuntimeContext.getTaskDataRangeRule();
            String taskItemInfo = taskItemExecuteRuntimeContext.getTaskItemInfo();
            //
            String taskProperties = taskItemExecuteRuntimeContext.getTaskProperties();
            String taskInstanceProperties = taskItemExecuteRuntimeContext.getTaskInstanceProperties();
            String taskItemIndexProperties = taskItemExecuteRuntimeContext.getTaskItemIndexProperties();
            //
            String threadName = taskItemExecuteRuntimeContext.createThreadName();
            taskItemExecuteRuntimeContext.setThreadName(threadName);
            //

            //Map<Integer, String> taskItemIndexToTaskItemParameterMap = taskExecuteContext.getTaskItemIndexToTaskItemParameterMap();
            //String taskItemParameter = taskItemIndexToTaskItemParameterMap.get(taskStage$type$index);


            //
            TaskItemExecute taskItemExecute = new TaskItemExecute(
                    taskExecutor, //
                    taskId, //
                    taskInstanceId,//
                    taskStage,//
                    taskStage$type,//
                    taskStage$type$index,//
                    //
                    getTaskExecuteContext().getTaskConfiguration().getTaskBaseInfo().getXid(), //
                    //
                    taskDataRangeRule,//
                    taskItemInfo,//
                    //
                    taskProperties,//
                    taskInstanceProperties,//
                    taskItemIndexProperties,//
                    secondAfterTaskStopCommandInstructed);//
            taskItemExecute.bind(taskItemExecuteRuntimeContext);
            taskItemExecuteList.add(taskItemExecute);
            //
            TaskItemExecuteTask taskItemExecuteTask = new TaskItemExecuteTask(taskItemExecute);
            taskItemExecuteTask.setThreadName(threadName);
            taskItemExecuteTaskList.add(taskItemExecuteTask);

        }
        for (TaskItemExecuteTask taskItemExecuteTask : taskItemExecuteTaskList) {
            String threadName = taskItemExecuteTask.getThreadName();
            TaskItemExecute taskItemExecute = taskItemExecuteTask.getTaskItemExecute();

            taskItemExecute.setThreadName(threadName);
            TaskItemExecuteThread taskItemExecuteThread = new TaskItemExecuteThread(taskItemExecuteTask);
            taskItemExecuteThread.setDaemon(true);
            taskItemExecuteThread.setName(threadName);
            taskItemExecuteThread.start();
            //add id
            if (timeoutMinutes != null) {
                taskExecutor.getExecutor().getTaskItemExecuteTimeoutTaskScheduledThreadPoolExecutor().schedule(new TaskItemExecuteTimeoutTask(threadName), timeoutMinutes, TimeUnit.MINUTES);
            }

        }
    }


    public void stop() {
        stopped = true;
    }

    //forceStop() 强制停止任务
    public void forceStop() {
        forceStop(taskId);
    }

    public void forceStop(String taskId) {
        forceStopped = true;
        TasksExecuteRuntimeContext.ergodicTaskExecuteRuntimeContextAndHandleBusiness(taskId, new TasksExecuteRuntimeContext.Business() {
            @Override
            public void doBusiness(ConcurrentHashMap<String, TaskItemExecuteRuntimeContext> taskItemIndexToTaskItemExecuteRuntimeContextMap, TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext) {
                forceStop(taskItemExecuteRuntimeContext);
            }
        });
    }


    private void forceStop(TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext) {
        taskItemExecuteRuntimeContext.needStop = true;//
        //检查当前任务分片是否完成
        if (!taskItemExecuteRuntimeContext.isFinished()) {
            Thread thread = Threads.getThreadByhreadName(taskItemExecuteRuntimeContext.getThreadName());
            if (thread == null) {
                return;
            }
            try {
                TaskItemExecuteThread taskItemExecuteThread = (TaskItemExecuteThread) thread;
                TaskItemExecuteTask taskItemExecuteTask = taskItemExecuteThread.getTaskItemExecuteTask();
                if (taskItemExecuteTask == null) {
                    return;
                }
                TaskItemExecute taskItemExecute = taskItemExecuteTask.getTaskItemExecute();
                if (taskItemExecute == null) {
                    return;
                }
                if (taskItemExecute.setForceStopWhenStatusInit()) {
                    log.warn("Force stop job, taskId:{}, item:{}", taskId, JSON.toJSONString(taskItemExecute));
                    new Thread(new Runnable() {
                        @Override
                        public void run() {
                            TaskItemExecuteTaskKill.kill(taskItemExecuteTask);
                        }
                    }).start();
                }
            } catch (Exception e) {
                log.info(e.getMessage(), e);
            }
            return;
        } else {
            return;
        }
    }


}
